/*
 * Singularity Engine
 * Copyright (C) 2010 Vyacheslav Vorobyov, All rights reserved.
 * See LICENSE for full license information.
 */
package org.singularity.io.util;

import java.io.BufferedInputStream;
import java.io.DataInputStream;
import java.io.EOFException;
import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.lang.reflect.Array;
import java.util.zip.GZIPInputStream;

import org.singularity.io.CloseableIterator;

/**
 * The counterpart utility to deserialize tuples (arrays) 
 * of fixed arity from binary file.
 * 
 * @author vjache
 * 
 * @see TupleSerializerPipe
 */
public abstract class TupleDeserializerIterator<T> implements
		CloseableIterator<T[]> {

	protected final DataInputStream _in;
	private final int _tupleSize;

	private T[] _tuple;

	/**
	 * @throws IOException
	 * 
	 */
	public TupleDeserializerIterator(File aFile) throws IOException {
		final InputStream fis = new BufferedInputStream(new FileInputStream(
				aFile), 64 * 1024);
		boolean zipped = fis.read() == 0 ? false : true;
		_in = new DataInputStream(zipped ? new GZIPInputStream(fis) : fis);
		_tupleSize = _in.readInt();
//		_tuple = new Object[_tupleSize];

		advance();
	}

	@SuppressWarnings("unchecked")
	private void advance() {
		try {
			int start = (int) _in.readByte();

			for (int i = start; i < _tupleSize; i++) {
				final T val = deserialize();
				if(_tuple==null)
				{
					_tuple = (T[]) Array.newInstance(val.getClass(),_tupleSize);
				}
				_tuple[i] = val;
			}

		} catch (EOFException e) {
			_tuple = null;
		} catch (IOException e) {
			throw new RuntimeException(e);
		}
	}

	@Override
	public void close() throws IOException {
		_in.close();
	}

	@Override
	public boolean hasNext() {
		return _tuple != null;
	}

	@Override
	public T[] next() {
		if (!hasNext())
			return null;
		final T[] ret = _tuple;
		advance();
		return ret;
	}

	@Override
	public void remove() {}

	/**
	 * Deserializes a value from internal data input stream. 
	 * @return
	 * @throws IOException
	 */
	protected abstract T deserialize() throws IOException;

}
